package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_01_user extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_01_user().init(
                10002,
                2,
                "Dwd_01_user"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //读取ods_db
        readOdsDb(tEnv,"Dwd_01_user");
        //从ods_db过滤出 用户注册的表数据
        Table userInfoTable = tEnv.sqlQuery(" select " +
                                                " data['id'] id," +
                                                " data['login_name'] login_name," +
                                                " data['nick_name'] nick_name," +
                                                " data['real_name'] real_name," +
                                                " data['user_level'] user_level," +
                                                " data['birthday'] birthday," +
                                                " data['gender'] gender," +
                                                " data['create_time'] create_time," +
                                                " data['operate_time'] operate_time," +
                                                " ts" +
                                                " from ods_db" +
                                                " where `database`='gmall2022'" +
                                                " and `table`='user_info'" +
                                                " and `type`='insert'");
        tEnv.createTemporaryView("user_info",userInfoTable);

        //写到kafka
        tEnv.executeSql(" create table dwd_user(" +
                           " id string," +
                           " login_name string," +
                           " nick_name string," +
                           " real_name string," +
                           " user_level string," +
                           " birthday string," +
                           " gender string," +
                           " create_time string," +
                           " operate_time string," +
                           " ts bigint" +
                           " )" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_USER));
        userInfoTable.executeInsert("dwd_user");

    }
}
